﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Threading;
using System.Net.Sockets;
using GeneralDFS.BusinessLogical.Config;
using GeneralDFS.BusinessLogical.Nodes;
using GeneralDFS.Common;
using GeneralDFS.DataEntity.IO;
using GeneralDFS.DataEntity.Config;
using GeneralDFS.BusinessLogical;
using GeneralDFS.BusinessLogical.IO;
using GeneralDFS.BusinessLogical.RemoteObject;

namespace GeneralDFS.BusinessLogical.Network
{
    public class AsyncDataNodeTcpListener
    {
        public AsyncDataNodeTcpListener() { }
        private readonly TcpListener _tcpListener;
        private DataNodeConfig dataNodeConfig;
        private DataNode dataNode;
        private DFSConfig config;
        private FsDir fsDir;

        public AsyncDataNodeTcpListener(DFSConfig config,IPAddress address,int port)
        {
            this.config = config;
            _tcpListener = new TcpListener(address,port);
        }

        public void ChangeToSelf(DataNode dataNode,DataNodeConfig dataNodeConfig)
        {
            this.dataNode = dataNode;
            this.dataNodeConfig = dataNodeConfig;
        }

        public void Read()
        {
            _tcpListener.Start();
            var tcpClient = _tcpListener.AcceptTcpClient();
            var networkStream = tcpClient.GetStream();
            var state = new CustomTcpState() { Guid = Guid.NewGuid(), NetworkStream = networkStream };
            ReadMetaData(ref state); //读取元数据
            state.PreBuffer = new byte[config.DetailConfig.BuffSize];
            state.Buffer = new byte[config.DetailConfig.BuffSize];
            state.PreBufferSize = state.PreBuffer.Length;
            state.BufferSize = state.Buffer.Length;
            fsDir = FSVolume.GetInstance().GetFreeDir(state.Guid);
            state.FileStream = Block.CreateBlockFile(state.Guid, fsDir);
            state.FileStream.Seek(0, SeekOrigin.Begin);
            state.Position = 0;
            state.PreEvent = new ManualResetEvent(false);
            state.PreEvent.Reset();
            state.AllDone = new ManualResetEvent(false);
            state.AllDone.Reset();
            LogUtil.LogInfo("=======Begin Read BlockData From NetworkStream！=====");
            networkStream.BeginRead(state.Buffer, 0, state.BufferSize, PreReadDataCallBack, state);
            state.AllDone.WaitOne();
        }

        /// <summary>
        /// 读取Block数据
        /// </summary>
        /// <param name="result"></param>
        public void PreReadDataCallBack(IAsyncResult result)
        {
            var state = result.AsyncState as CustomTcpState;
            state.AllDone.Set();
            var networkStream = state.NetworkStream;
            if (networkStream == null||null==state.FileStream) return;
            try
            {
                var recv = state.NetworkStream.EndRead(result);
                if (state.Position != 0)
                {
                    state.PreEvent.WaitOne();
                    state.PreEvent.Reset();
                    Array.Copy(state.PreBuffer, 0, state.Buffer, 0, recv);
                }
                state.FileStream.BeginWrite(state.Buffer, 0, recv, WriteCallBack, state);
                //state.FileStream.Write(state.Buffer, 0, recv);
                //state.FileStream.Flush(true);
                state.Position += recv;
                if(state.Position==state.FileBlockMetaData.Length)
                {
                    state.PreEvent.WaitOne();
                    state.PreEvent.Reset();
                    state.FileStream.Flush(true);
                    state.FileStream.Close();
                    LogUtil.LogInfo("FileStream Closed!");
                    //fsDir.MoveBlockToCurrentDirectory(state.Guid);
                    LogUtil.LogInfo("Move block:{0} to current Dir",state.Guid);
                    networkStream.Flush();
                    networkStream.Close();
                    LogUtil.LogInfo("NetworkStream Closed!");
                    LogUtil.LogInfo("=======End Read BlockData From NetworkStream！=====");
                    LogUtil.LogInfo(string.Format("=======Update FileBlockInfo at Path:{0}=====", state.FileBlockMetaData.FullPath));
                    dataNode.NameNodeRemoteObject.CheckFIleBlockContainsDataNode(state.FileBlockMetaData.FullPath, state.FileBlockMetaData,dataNode.DataNodeInfo);
                }
                else
                {
                    networkStream.BeginRead(state.PreBuffer, 0, state.PreBufferSize, PreReadDataCallBack, state);
                    state.AllDone.WaitOne();
                }
            }
            catch (Exception ex)
            {
                LogUtil.LogError(ex);
            }
        }

        private void WriteCallBack(IAsyncResult result)
        {
            CustomTcpState state = result.AsyncState as CustomTcpState;
            Debug.Assert(state != null, "state is null");
            state.FileStream.EndWrite(result);
            state.PreEvent.Set();
        }

        /// <summary>
        /// 读取元数据
        /// </summary>
        /// <param name="state"></param>
        public void ReadMetaData(ref CustomTcpState state)
        {
            LogUtil.LogInfo("=====Begin to read MetaData=====");
            var stream = state.NetworkStream;
            var metaDataLengthBytes = new byte[sizeof(int)];
            int length = stream.Read(metaDataLengthBytes, 0, metaDataLengthBytes.Length);
            int metaDataLength = BitConverter.ToInt32(metaDataLengthBytes, 0);
            LogUtil.LogInfo(String.Format("=====MetaData Length:{0}",length));
            var metaDataStream = new MemoryStream();
            metaDataStream.Seek(0, SeekOrigin.Begin);
            int count = 0;
            var buffer = new byte[config.DetailConfig.BuffSize];
            try
            {
                while (count < metaDataLength)
                {
                    if (metaDataLength - count < config.DetailConfig.BuffSize)
                    {
                        length = stream.Read(buffer, 0, metaDataLength - count);
                        metaDataStream.Write(buffer, 0, length);
                        break;
                    }
                    else
                    {
                        length = stream.Read(buffer, 0, buffer.Length);
                        metaDataStream.Write(buffer, 0, length);
                        count += length;
                    }
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
                LogUtil.LogError(ex);
            }

            state.FileBlockMetaData = SerializeHelper.GetModel(metaDataStream) as FileBlockMetaData;
            Debug.Assert(state.FileBlockMetaData != null, "FileBlockMetaData != null");
            state.FileBlockMetaData.Guid = state.Guid;
            state.FileBlockMetaData.Hash = state.Guid.ToString();
            Block.SaveMetaData(fsDir,state.FileBlockMetaData, state.Guid);
            LogUtil.LogInfo(String.Format("FileName:{0} Guid:{1} Offset:{2} Length:{3}", state.FileBlockMetaData.FileName, state.Guid.ToString(), state.FileBlockMetaData.Offset, state.FileBlockMetaData.Length));
            LogUtil.LogInfo("=====End read MetaData=====");
        }
    }
}